package cn.iotab.iotkit.transport.mqtt.handler;

import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionInfo;
import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionReportInfo;
import cn.iotab.iotkit.transport.api.common.queue.SendResultCallback;
import cn.iotab.iotkit.transport.api.driver.DriverSession;
import cn.iotab.iotkit.transport.mqtt.MqttTransportContext;
import cn.iotab.iotkit.transport.mqtt.session.MqttSessionCtx;
import cn.iotab.iotkit.transport.mqtt.session.MqttTopicMatcher;
import cn.iotab.iotkit.transport.mqtt.topic.MqttTopics;
import cn.iotab.iotkit.transport.mqtt.topic.TopicMeta;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttSubAckMessage;
import io.netty.handler.codec.mqtt.MqttSubAckPayload;
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

@Slf4j
public class SubscribeHandler extends AbstractMqttMessageHandler<MqttSubscribeMessage> {

    public SubscribeHandler(MqttTransportContext context, MqttSessionCtx deviceSessionCtx) {
        super(context, deviceSessionCtx);
    }

    @Override
    public void process(ChannelHandlerContext ctx, MqttSubscribeMessage mqttMsg) {
        if (!checkConnected(ctx, mqttMsg)) {
            return;
        }
        //定义topic列表
        List<String> topics = mqttMsg.payload().topicSubscriptions().stream().map(MqttTopicSubscription::topicName).collect(Collectors.toList());
        if (log.isTraceEnabled()) {
            log.trace("[{}] Processing subscription [{}]!", sessionId, Arrays.toString(topics.toArray()));
        }

        DriverSession sessionMD = deviceSessionCtx.getDriverSession();
        // 这一步不肯能发生，判断预防一下
        if (sessionMD == null) {
            log.error("[{}] sessionMetaData is null while subscribe topics {}", sessionId, Arrays.toString(topics.toArray()));
            ctx.writeAndFlush(createAllSubAckFailedMessage(mqttMsg.variableHeader().messageId(), topics.size()));
            return;
        }
        SessionInfo sessionInfo = sessionMD.getSessionInfo();
        //
        boolean subscribeMsgTopic = false;
        boolean subscribeReplyTopic = false;
        //
        List<Integer> grantedQoSList = new ArrayList<>();
        for (MqttTopicSubscription subscription : mqttMsg.payload().topicSubscriptions()) {
            String topic = subscription.topicName();
            MqttQoS reqQoS = subscription.qualityOfService();
            TopicMeta topicMeta = MqttTopics.parseTopic(topic);
            //校验productKey与deviceID合法性
            if (topicMeta.isValid()) {
                String productKey = topicMeta.getProductKey();
                String deviceId = topicMeta.getDeviceId();
                if (!StringUtils.equals(productKey, sessionInfo.getProductKey())
                        || !StringUtils.equals(deviceId, sessionInfo.getDeviceId())) {
                    log.error("[{}] Failed to subscribe to [{}][{}], please see correct productKey:{}, deviceId:{}",
                            sessionId, topic, reqQoS, sessionInfo.getProductKey(), sessionInfo.getDeviceId());
                    grantedQoSList.add(MqttQoS.FAILURE.value());
                    continue;
                }
            }
            //校验topic类型
            switch (topicMeta.getTopicType()) {
                case PLATFORM_POST_DEVICE:
                    registerSubQoS(topic, grantedQoSList, reqQoS);
                    subscribeMsgTopic = true;
                    break;
                case PLATFORM_REPLY_DEVICE:
                    registerSubQoS(topic, grantedQoSList, reqQoS);
                    subscribeReplyTopic = true;
                    break;
                default:
                    log.warn("[{}] Failed to subscribe to [{}][{}]: invalid topic", sessionId, topic, reqQoS);
                    grantedQoSList.add(MqttQoS.FAILURE.value());
            }
        }
        //
        transportDriverService.sessionActivity(sessionInfo);

        //订阅结果更新并上报
        if (subscribeMsgTopic || subscribeReplyTopic) {
            if (!sessionMD.isSubscribedMessageTopic() && subscribeMsgTopic) {
                sessionMD.setSubscribedMessageTopic(true);
            }
            if (!sessionMD.isSubscribedReplyTopic() && subscribeReplyTopic) {
                sessionMD.setSubscribedReplyTopic(true);
            }
            //
            long updateTime = System.currentTimeMillis();
            SessionReportInfo reportInfo = SessionReportInfo.builder()
                    .subscribedMessageTopic(sessionMD.isSubscribedMessageTopic())
                    .subscribedReplyTopic(sessionMD.isSubscribedReplyTopic())
                    .lastActivityTime(updateTime).build();

            transportDriverService.reportSessionInfo(deviceSessionCtx.getSessionInfo(), reportInfo, new SendResultCallback<Void>() {
                @Override
                public void onSuccess(Void msg) {
                    sessionMD.setLastReportedActivityTime(updateTime);
                }

                @Override
                public void onError(Throwable e) {
                    log.warn("[{}] Failed to report last activity time", sessionId, e);
                }
            });
        }

        ctx.writeAndFlush(createSubAckMessage(mqttMsg.variableHeader().messageId(), grantedQoSList));
    }

    private void registerSubQoS(String topic, List<Integer> grantedQoSList, MqttQoS reqQoS) {
        grantedQoSList.add(getMinSupportedQos(reqQoS));
        deviceSessionCtx.getMqttQoSMap().put(new MqttTopicMatcher(topic), getMinSupportedQos(reqQoS));
    }

    private static MqttSubAckMessage createAllSubAckFailedMessage(Integer msgId, int subNum) {
        List<Integer> grantedQoSList = new ArrayList<>();
        for (int i = 0; i < subNum; i++) {
            grantedQoSList.add(MqttQoS.FAILURE.value());
        }
        return createSubAckMessage(msgId, grantedQoSList);
    }

    private static MqttSubAckMessage createSubAckMessage(Integer msgId, List<Integer> grantedQoSList) {
        MqttFixedHeader mqttFixedHeader =
                new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttMessageIdVariableHeader mqttMessageIdVariableHeader = MqttMessageIdVariableHeader.from(msgId);
        MqttSubAckPayload mqttSubAckPayload = new MqttSubAckPayload(grantedQoSList);
        return new MqttSubAckMessage(mqttFixedHeader, mqttMessageIdVariableHeader, mqttSubAckPayload);
    }
}
